package com.zetyun.tiger.format.header;

import com.zetyun.rt.meta.annotation.ActionMeta;
import com.zetyun.rt.sdk.action.MapAction;
import com.zetyun.rt.sdk.model.RtEvent;
import com.zetyun.rt.sdk.operator.OperatorContext;


@ActionMeta(
        id = "tiger.header-to-body",
        tags = {"formatter"},
        category = "map",
        name = "ExtractTopicAction",
        parameterClass = ExtractTopicParameter.class,
        description = "根据特定规则组装Sink主题信息"
)
public class ExtractTopicAction extends MapAction {

    private ExtractTopicParameter parameters;

    @Override
    public void init(OperatorContext context) {
        super.init(context);
        parameters = context.getParameters();

    }

    @Override
    public RtEvent apply(RtEvent rtEvent) throws Exception {

        String topic = rtEvent.getHeader(parameters.getTopicField(), String.class);
        String dbName = "";
        String tbName = "";
        if (topic != null) {
            String[] arr = topic.split(parameters.getTopicSeparator());
            dbName = arr[parameters.getDbIndex()];
            tbName = arr[parameters.getTbIndex()];
        }

        String destTopicName = "ODS_" + dbName + "_" + tbName;
        rtEvent.setHeader("SINK_TOPIC", destTopicName);

        // customer
//        rtEvent.setValue("SINK_TOPIC", destTopicName);
//        rtEvent.setValue("header", rtEvent.getHeaders());


        return rtEvent;
    }
}
